package com.ydlclass.proxy.handler;

import com.ydlclass.NettyBootstrapInitializer;
import com.ydlclass.Transport.message.Request.RequestPayload;
import com.ydlclass.Transport.message.Request.YrpcRequest;
import com.ydlclass.YrpcBootstrap;
import com.ydlclass.annotation.TryTimes;
import com.ydlclass.compress.CompressFactory;
import com.ydlclass.discovery.Registry;
import com.ydlclass.enumeration.RequestTypeEnum;
import com.ydlclass.exceptions.DiscoveryException;
import com.ydlclass.exceptions.NetworkException;
import com.ydlclass.serialize.SerializerFactory;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 使用动态代理来生成代理对象的invoke方法
 * 封装了客户端通信的基础逻辑，每一个代理对象的调用方法都是通过此对象来完成的invoke方法
 * 1、 发现可用服务
 * 2、 看map中是否有对应的实例，如果没有则调用注册中心注册服务
 * 3、 进行netty通信
 * @Author duWenJian
 * @Date 2024/2/19 20:32
 * @Version 1.0
 */
@Slf4j
@Data
public class RpcConsumerInvocationHandler<T> implements InvocationHandler {
    /**
     * 此处需要一个注册中心和一个接口
     */
    private final Class<T> Interface;

    //private final Registry registry;

    private final String group;


    /**
     * 所有的方法调用本质都会走到这里
     * @param proxy
     * @param method
     * @param args
     * @return
     * @throws Throwable
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        log.info("调用方法：" + method.getName());
        log.info("调用参数：" + args);

        // todo 封装报文
        //  需要各种请求id,类型进行处理

        // 默认值: 0代表不重试
        int tryTimes = 0;
        int intervalTime = 0;
        // 从接口中判断是否需要重试
        TryTimes annotation = method.getAnnotation(TryTimes.class);
        if (annotation != null) {
            tryTimes = annotation.tryTimes();
            intervalTime = annotation.intervalTime();
        }

        while (true) {
            try {

                long RequestId = YrpcBootstrap.getInstance().getConfiguration().getIdGenerator().getId();

                YrpcRequest yrpcRequest = YrpcRequest.builder()
                        .requestId(RequestId)
                        .requestType(RequestTypeEnum.NormalRequest.getId())
                        .compressType(CompressFactory.getCompress(YrpcBootstrap.getInstance().getConfiguration().getCompressionType()).getCode())
                        .timeStamp(System.currentTimeMillis())
                        .serializeType(SerializerFactory.getSerializer(YrpcBootstrap.getInstance().getConfiguration().getSerializerType()).getCode())
                        .requestPayload(RequestPayload.builder()
                                .interfaceName(Interface.getName())
                                .methodName(method.getName())
                                .parameterTypes(method.getParameterTypes())
                                .parametersValue(args)
                                .returnType(method.getReturnType())
                                .build()
                        )
                        .build();
                // 将请求存入本地线程，需要在合适的时候remove
                YrpcBootstrap.REQUEST_THREAD_LOCAL.set(yrpcRequest);


                // 1、 lookup() 寻找可用的服务实例， 返回一个ip+port
                // 获取当前配置的负载均衡器，选取一个可用节点
                InetSocketAddress address = YrpcBootstrap.getInstance().getConfiguration().getLoadBalance().selectServiceAddress(Interface.getName(), group);
                if (log.isDebugEnabled()) {
                    log.debug("服务调用方发现了:{}的可用主机:{},{}", Interface.getName(), address.getAddress(), address.getPort());
                }


                //   使用netty连接到服务器，发送 调用服务的 名称+方法名称+参数列表，返回结果
                // q: 整个连接过程放在这里行不行，也就意味着每次调用都会产生一个新的netty连接，如何缓存我们的连接
                //    也就意味着每次在此处建立一个新的连接是不合适的。
                // 解决方案: 如何缓存我们的连接，如何复用我们的连接
                //         尝试从缓存中获取一个可用的连接，如果没有可用的连接，就新建一个连接

                // 2、尝试从缓存中获取channel，没有就新建
                Channel channel = getAvailableChannel(address);

                if (log.isDebugEnabled()) {
                    log.debug("服务调用方发现可用主机:{},{}", address.getHostString(), address.getPort());
                }

                // ------------------------------------异步策略----------------------------------
                CompletableFuture<Object> future = new CompletableFuture<>();
                //  需要将CompletableFuture暴露出去，
                YrpcBootstrap.PENDING_REQUEST.put(yrpcRequest.getRequestId(), future);

                // 这里直接writeAndFlush发出一个请求，请求实例就会进入pipeline，执行出站操作
                // 我们可以对里面的handler进行处理request,  将request转化为  --- 二进制报文
                channel.writeAndFlush(yrpcRequest).addListener((ChannelFutureListener) promise -> {
                    // 当前的promise将来返回的结果是什么，writeAndFlush的返回结果是什么？
                    // 一旦数据被写出去，这个个promise也就随之结束
                    // 但是我们想要的是 : 服务端给我们的响应结果，所以使用这个CompletableFuture有问题
                    // 是不是应该将CompletableFuture挂起并暴露出去，在得到服务端的响应的时候调用complete方法
                    if (!promise.isSuccess()) {
                        future.completeExceptionally(promise.cause());
                    }
                });

                YrpcBootstrap.REQUEST_THREAD_LOCAL.remove();

                // todo 如果没有地方处理这个CompletableFuture，这里就会阻塞，等待completeExceptionally抛出异常
                //  q: 我们需要在哪里调用complete方法来获取服务端的响应;;;;  谁发来的我找谁，肯定client接受是channelRead的时候
                // 5、获得响应的结果

                return YrpcBootstrap.PENDING_REQUEST.get(yrpcRequest.getRequestId()).get(3, TimeUnit.SECONDS);
            } catch (Exception e) {
                // 次数减一，兵器等在固定时间，固定时间有一定的问题，重试风暴
                tryTimes--;
                try {
                  Thread.sleep(intervalTime*(3-tryTimes));
                } catch (InterruptedException e1){
                    log.error("再重试时发生异常:{}",e1);
                }
                if (tryTimes < 0){
                    log.error("调用方法{}失败，重试次数:{},仍不可调用:{}" , method.getName(),tryTimes,e);
                    break;
                }
                log.error("在进行第:{}次重试的时候发生异常",(3-tryTimes));
            }
        }
        throw new RuntimeException("执行方法"+method.getName()+"调用失败");
    }

    public RpcConsumerInvocationHandler(Class<T> anInterface, Registry registry ,String group) {
        Interface = anInterface;
        //this.registry = registry;
        this.group = group;
    }


    /**
     * 获取一个可用的channel
     * @param address
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws TimeoutException
     */
    private Channel getAvailableChannel(InetSocketAddress address) {

        // 1、尝试从缓存中获取
        Channel channel = YrpcBootstrap.CHANNEL_CACHE.get(address);

        // 2、拿不到就去建立简介
        if (channel == null ){
            // sync会阻塞等待连接的建立。netty还提供了异步的连接方式
            // await会进行阻塞当前线程获取返回值，连接的过程是异步的，发送的过程是异步的，netty大多数操作都是异步的
            // 如果发生了异常，sync会主动在这里抛出异常，await会在子线程抛出异常

            /**
             *-----------------------------------异步策略----------------------------------
             * 异步的方式来进行,connect连接好了之后，会调用监听器里面的回调方法
             *
             */
            CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
            NettyBootstrapInitializer.getBootstrap().connect(address).addListener( (ChannelFutureListener) promise -> {
                if (promise.isDone()) {
                    if (log.isDebugEnabled()) {
                        log.debug("已经和：{}成功建立连接。", address);
                    }
                    completableFuture.complete(promise.channel());
                } else if (!promise.isSuccess()){
                    completableFuture.completeExceptionally(promise.cause());
                }
            });

            // get方法是阻塞等待
            try {
                channel = completableFuture.get(3, TimeUnit.SECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                log.error("获取channel通道发生异常:{}",e);
                throw new DiscoveryException(e);
            }
            // 缓存
            YrpcBootstrap.CHANNEL_CACHE.put(address,channel);

        }

        if (channel == null ){
            log.error("获取建立{}通道时发生异常", address);
            throw new NetworkException("建立通道失败！！");
        }
        return channel;
    }
}
